/*
    Muduo库封装
    1. 缓冲区
    2. 协议
    3. 连接
    4. 服务端
    5. 客户端
*/

#pragma once

#include "abstract.hpp"
#include "detail.hpp"
#include "fields.hpp"
#include "message.hpp"
#include <muduo/net/Buffer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpServer.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/TcpClient.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/net/EventLoopThread.h>
#include <mutex>

namespace myrpc
{
    //muduobuffer封装
    class MuduoBuffer : public BaseBuffer
    {
    public:
        typedef std::shared_ptr<MuduoBuffer> ptr;

        MuduoBuffer(muduo::net::Buffer* buf) :_buf(buf) {}
        virtual size_t readableSize() override { return _buf->readableBytes(); } // 缓冲区中可供读取的字节数
        virtual int32_t peekInt32() override { return _buf->peekInt32(); } // 观察前4字节数据
        virtual void retrieveInt32() override { _buf->retrieveInt32(); } // 读走前4字节数据
        virtual int32_t readInt32() override { return _buf->readInt32(); } // 读取前4字节数据
        virtual std::string retrieveAsString(size_t len) { return _buf->retrieveAsString(len); } // 读取指定长度的数据
    private:
        muduo::net::Buffer* _buf;
    };

    class BufferFactory
    {
    public:
        template<class ...Args>
        static BaseBuffer::ptr create(Args&& ...args)
        {
            return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
        }
    };

    //应用层协议
    //|--len(4)--|--Mtype(4)--|--IDLen(4)--|--ID--|--body--|
    class LVProtocol : public BaseProtocol 
    {
    public:
        typedef std::shared_ptr<LVProtocol> ptr;

        //判断buff中的数据量是否满足完整的一条消息
        virtual bool canProcessed(const BaseBuffer::ptr &buf) override
        {
            if(buf->readableSize() < len_Length)
            {
                return false;
            }
            int32_t total_len = buf->peekInt32();
            if(total_len + len_Length > buf->readableSize())
            {
                return false;
            }
            return true;
        }

        //获取消息并进行反序列化
        virtual bool onMessage(const BaseBuffer::ptr &buf, BaseMessage::ptr &msg) override
        {
            int32_t total_len = buf->readInt32();
            MType mtype = (MType)buf->readInt32();
            int32_t idlen = buf->readInt32();
            std::string id = buf->retrieveAsString(idlen);
            int32_t bodylen = total_len - Mtype_Length - IDLen_Length - idlen;
            std::string body = buf->retrieveAsString(bodylen);

            msg = MessageFactory::create(mtype);
            if(msg.get() == nullptr)
            {
                ELOG("消息类型错误，构造失败");
                return false;
            }
            msg->unserialize(body);
            msg->setId(id);
            msg->setMType(mtype);
            return true;
        }

        //将消息进行序列化
        virtual std::string serialize(const BaseMessage::ptr &msg)
        {
            std::string body = msg->serialize();
            std::string id = msg->rid();
            int32_t h_total_len = Mtype_Length + IDLen_Length + id.size() + body.size();
            int32_t n_total_len = htonl(h_total_len);
            int32_t idlen = htonl(id.size());
            auto mtype = htonl((int32_t)msg->mtype());

            std::string ret;
            ret.reserve(h_total_len + len_Length);
            ret.append((const char*)&n_total_len, len_Length);
            ret.append((const char*)&mtype, Mtype_Length);
            ret.append((const char*)&idlen, IDLen_Length);
            ret.append(id);
            ret.append(body);
            return ret;
        }
    private:
        const static size_t len_Length = 4;
        const static size_t Mtype_Length = 4;
        const static size_t IDLen_Length = 4;
    };

    class ProtocolFactory
    {
    public:
        template<class ...Args>
        static BaseProtocol::ptr create(Args&& ...args)
        {
            return std::make_shared<LVProtocol>(std::forward<Args>(args)...);
        }
    };

    //连接
    class MuduoConnection : public BaseConnection 
    {
    public:
        typedef std::shared_ptr<MuduoConnection> ptr;

        MuduoConnection(const BaseProtocol::ptr& protocol, const muduo::net::TcpConnectionPtr& conn)
            :_conn(conn), _protocol(protocol) {}

        virtual void send(const BaseMessage::ptr &msg) override
        {
            std::string message = _protocol->serialize(msg);
            _conn->send(message);
        }
        virtual void shutdown() override
        {
            _conn->shutdown();
        }
        virtual bool connected() override
        {
            return _conn->connected();
        }
    private:
        BaseProtocol::ptr _protocol;
        muduo::net::TcpConnectionPtr _conn;
    };

    class ConnectionFactory
    {
    public:
        template<class ...Args>
        static BaseConnection::ptr create(Args&& ...args)
        {
            return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
        }
    };

    //服务端
    class MuduoServer : public BaseServer 
    {
    public:
        using ptr = std::shared_ptr<MuduoServer>;
        MuduoServer(int port)
            :_server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), 
            "MuduoServer", muduo::net::TcpServer::kReusePort)
            , _protocol(ProtocolFactory::create())
        {}
        virtual void start() override
        {
            //设置事件回调
            _server.setConnectionCallback(std::bind(&MuduoServer::onConnection, this, std::placeholders::_1));
            _server.setMessageCallback(
                std::bind(&MuduoServer::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            
            _server.start(); // 开始监听
            _baseloop.loop(); // 死循环事件监控
        }
    private:
        void onConnection(const muduo::net::TcpConnectionPtr& conn)
        {
            if(conn->connected())
            {
                ILOG("连接建立");
                auto muduo_conn = ConnectionFactory::create(_protocol, conn);
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _conns.insert({conn, muduo_conn});
                }
                if(_cb_connection) _cb_connection(muduo_conn);
            }
            else
            {
                ILOG("连接断开");
                BaseConnection::ptr muduo_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if(it == _conns.end())
                    {
                        return;
                    }
                    muduo_conn = it->second;
                    _conns.erase(it);
                }
                if(_cb_close) _cb_close(muduo_conn);
            }
        }

        void onMessage(const muduo::net::TcpConnectionPtr& conn, muduo::net::Buffer* buff, muduo::Timestamp)
        {
            auto base_buf = BufferFactory::create(buff);
            while(true)
            {
                if(!_protocol->canProcessed(base_buf))
                {
                    if(base_buf->readableSize() > maxSize)
                    {
                        conn->shutdown();
                        ELOG("缓冲区中数据过大");
                        return;
                    }
                    ELOG("数据量不足");
                    break;
                }

                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(base_buf, msg);
                if(!ret)
                {
                    conn->shutdown();
                    ELOG("缓冲区中数据错误");
                    return;
                }

                BaseConnection::ptr base_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if(it == _conns.end())
                    {
                        conn->shutdown();
                        return;
                    }
                    base_conn = it->second;
                }
                if(_cb_message) _cb_message(base_conn, msg);
            }
        }

        const static size_t maxSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server;
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _conns;
        std::mutex _mutex;
    };

    class ServerFactory
    {
    public:
        template<class ...Args>
        static BaseServer::ptr create(Args&& ...args)
        {
            return std::make_shared<MuduoServer>(std::forward<Args>(args)...);
        }
    };

    //客户端
    class MuduoClient : public BaseClient 
    {
    public:
        using ptr = std::shared_ptr<MuduoClient>;

        MuduoClient(const std::string& sip, int port)
            :_protocol(ProtocolFactory::create())
            , _downlatch(1)
            , _baseloop(_loopthread.startLoop())
            , _client(_baseloop, muduo::net::InetAddress(sip, port), "MuduoClient")
        {}
        virtual void connect()
        {
            //设置事件回调
            _client.setConnectionCallback(std::bind(&MuduoClient::onConnection, this, std::placeholders::_1));
            _client.setMessageCallback(
                std::bind(&MuduoClient::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            //连接
            _client.connect();
            _downlatch.wait();
        }
        virtual void shutdown() override
        {
            return _client.disconnect();
        }
        virtual bool send(const BaseMessage::ptr& msg) override
        {
            if(!_conn->connected())
            {
                ELOG("连接已断开");
                return false;
            }
            _conn->send(msg);
            return true;
        }
        virtual BaseConnection::ptr connection() override
        {
            return _conn;
        }
        virtual bool connected() override
        {
            return (_conn && _conn->connected());
        }

    private:
        void onConnection(const muduo::net::TcpConnectionPtr& conn)
        {
            if(conn->connected())
            {
                ILOG("连接建立");
                _downlatch.countDown(); // 计数--，为0唤醒
                _conn = ConnectionFactory::create(_protocol, conn);
            }    
            else
            {
                ILOG("连接断开");
                _conn.reset();
            }
        }

        void onMessage(const muduo::net::TcpConnectionPtr& conn, muduo::net::Buffer* buff, muduo::Timestamp)
        {
            auto base_buf = BufferFactory::create(buff);
            while(true)
            {
                if(!_protocol->canProcessed(base_buf))
                {
                    if(base_buf->readableSize() > maxSize)
                    {
                        conn->shutdown();
                        ELOG("缓冲区中数据过大");
                        return;
                    }
                    break;
                }

                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(base_buf, msg);
                if(!ret)
                {
                    conn->shutdown();
                    ELOG("缓冲区中数据错误");
                    return;
                }

                if(_cb_message) _cb_message(_conn, msg);
            }
        }

        const static size_t maxSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        BaseConnection::ptr _conn;
        muduo::CountDownLatch _downlatch;
        muduo::net::EventLoopThread _loopthread;
        muduo::net::EventLoop* _baseloop = nullptr;
        muduo::net::TcpClient _client;
        std::mutex _mutex;
    };

    class ClientFactory
    {
    public:
        template<class ...Args>
        static BaseClient::ptr create(Args&& ...args)
        {
            return std::make_shared<MuduoClient>(std::forward<Args>(args)...);
        }
    };
}